package com.kqkj.dyrule.marketing.functions;

import com.kqkj.dyrule.marketing.beans.DynamicKeyedBean;
import com.kqkj.dyrule.marketing.beans.EventBean;
import com.kqkj.dyrule.marketing.beans.MarketingRule;
import com.kqkj.dyrule.marketing.utils.RuleSimulatorFromJson;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.lang.reflect.Field;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 方便后续动态keyby的数据复制ProcessFunction
 */
public class DynamicKeyByReplicateFunction extends ProcessFunction<EventBean, DynamicKeyedBean> {

    Set<String> keyByFieldsSet;
    @Override
    public void open(Configuration parameters) throws Exception {
        //获取系统中所有规则列表
        List<MarketingRule> ruleList = RuleSimulatorFromJson.getRule();
        //从规则列表中遍历每个规则，获取每个规则的keyby字段名，并放入set集合去重
        keyByFieldsSet = new HashSet<>();
        for (MarketingRule rule : ruleList) {
            keyByFieldsSet.add(rule.getKeyByFields());
        }

    }

    @Override
    public void processElement(EventBean eventBean, Context context, Collector<DynamicKeyedBean> out) throws Exception {


        for (String keyByField : keyByFieldsSet) {
            StringBuilder sb = new StringBuilder();
            String[] fieldNames = keyByField.split(",");
            //拼装keyByField中指定每一个字段的值
            for (String fieldName : fieldNames) {
                Class<?> beanClass = Class.forName("com.kqkj.dyrule.marketing.beans.EventBean");
                Field declaredField = beanClass.getDeclaredField(fieldName);
                declaredField.setAccessible(true);
                String fieldValue = (String)declaredField.get(eventBean);
                sb.append(fieldValue).append(",");
            }
            String keyByValue = sb.toString().substring(0, sb.length() - 1);
            eventBean.setKeyByValue(keyByValue);
            DynamicKeyedBean dynamicKeyedBean = new DynamicKeyedBean(keyByValue, keyByField, eventBean);
            out.collect(dynamicKeyedBean);
        }




    }
}
